//! # 消息总线测试
//!
//! 提供消息总线系统的单元测试和集成测试

use crate::prelude::*;
use serde_json::json;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use tokio::time::{timeout, Duration};

/// 测试帮助器
pub struct TestHelper;

impl TestHelper {
    /// 创建测试消息
    pub fn create_test_message(topic: &str, payload: serde_json::Value) -> Message {
        Message::new(topic, payload)
    }

    /// 创建测试事件
    pub fn create_test_event(topic: &str, user_id: u64, action: &str) -> Message {
        Message::event(
            topic,
            json!({
                "user_id": user_id,
                "action": action,
                "timestamp": chrono::Utc::now().to_rfc3339()
            })
        )
    }

    /// 等待条件满足
    pub async fn wait_for_condition<F>(condition: F, timeout_ms: u64) -> bool
    where
        F: Fn() -> bool,
    {
        let start = std::time::Instant::now();
        let timeout_duration = Duration::from_millis(timeout_ms);

        while start.elapsed() < timeout_duration {
            if condition() {
                return true;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }

        false
    }

    /// 收集订阅者接收的消息
    pub async fn collect_messages(
        subscriber: Box<dyn Subscriber>,
        expected_count: usize,
        timeout_ms: u64,
    ) -> Vec<Message> {
        let mut messages = Vec::new();
        let start = std::time::Instant::now();
        let timeout_duration = Duration::from_millis(timeout_ms);

        while messages.len() < expected_count && start.elapsed() < timeout_duration {
            if let Ok(Some(message)) = subscriber.receive_timeout(100).await {
                messages.push(message);
            }
        }

        messages
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_message_creation() {
        let message = TestHelper::create_test_message(
            "test.topic",
            json!({"key": "value"})
        );

        assert_eq!(message.topic(), "test.topic");
        assert_eq!(message.payload()["key"], "value");
    }

    #[tokio::test]
    async fn test_event_creation() {
        let event = TestHelper::create_test_event("user.events.login", 123, "login");
        
        assert_eq!(event.topic(), "user.events.login");
        assert_eq!(event.metadata.attributes.message_type, MessageType::Event);
        assert_eq!(event.payload()["user_id"], 123);
        assert_eq!(event.payload()["action"], "login");
    }

    #[tokio::test]
    async fn test_basic_publish_subscribe() {
        let bus = InMemoryMessageBus::new().await.unwrap();

        // 订阅消息
        let subscriber = bus.subscribe("test.topic").await.unwrap();

        // 发布消息
        let message = TestHelper::create_test_message(
            "test.topic",
            json!({"content": "hello world"})
        );
        bus.publish(message).await.unwrap();

        // 接收消息
        let received = subscriber.receive_timeout(1000).await.unwrap();
        assert!(received.is_some());
        assert_eq!(received.unwrap().payload()["content"], "hello world");
    }

    #[tokio::test]
    async fn test_wildcard_subscription() {
        let bus = InMemoryMessageBus::new().await.unwrap();

        // 通配符订阅
        let subscriber = bus.subscribe("user.events.*").await.unwrap();

        // 发布匹配的消息
        let message1 = TestHelper::create_test_event("user.events.login", 123, "login");
        let message2 = TestHelper::create_test_event("user.events.logout", 123, "logout");
        let message3 = TestHelper::create_test_event("user.profile.update", 123, "update");

        bus.publish(message1).await.unwrap();
        bus.publish(message2).await.unwrap();
        bus.publish(message3).await.unwrap(); // 这个不应该被接收

        // 收集消息
        let messages = TestHelper::collect_messages(subscriber, 2, 1000).await;
        assert_eq!(messages.len(), 2);
        
        let topics: Vec<&str> = messages.iter().map(|m| m.topic()).collect();
        assert!(topics.contains(&"user.events.login"));
        assert!(topics.contains(&"user.events.logout"));
        assert!(!topics.contains(&"user.profile.update"));
    }

    #[tokio::test]
    async fn test_multi_level_wildcard() {
        let bus = InMemoryMessageBus::new().await.unwrap();

        // 多级通配符订阅
        let subscriber = bus.subscribe("user.**").await.unwrap();

        // 发布不同级别的消息
        let messages = vec![
            TestHelper::create_test_message("user.events.login", json!({"id": 1})),
            TestHelper::create_test_message("user.profile.settings.theme", json!({"id": 2})),
            TestHelper::create_test_message("user.notifications", json!({"id": 3})),
            TestHelper::create_test_message("admin.events", json!({"id": 4})), // 不匹配
        ];

        for message in messages {
            bus.publish(message).await.unwrap();
        }

        // 应该收到前3个消息
        let received_messages = TestHelper::collect_messages(subscriber, 3, 1000).await;
        assert_eq!(received_messages.len(), 3);
    }

    #[tokio::test]
    async fn test_multiple_subscribers() {
        let bus = InMemoryMessageBus::new().await.unwrap();

        // 创建多个订阅者
        let subscriber1 = bus.subscribe("broadcast.news").await.unwrap();
        let subscriber2 = bus.subscribe("broadcast.news").await.unwrap();
        let subscriber3 = bus.subscribe("broadcast.*").await.unwrap();

        // 发布广播消息
        let message = TestHelper::create_test_message(
            "broadcast.news",
            json!({"headline": "重要新闻"})
        );
        bus.publish(message).await.unwrap();

        // 所有订阅者都应该收到消息
        let received1 = subscriber1.receive_timeout(500).await.unwrap();
        let received2 = subscriber2.receive_timeout(500).await.unwrap();
        let received3 = subscriber3.receive_timeout(500).await.unwrap();

        assert!(received1.is_some());
        assert!(received2.is_some());
        assert!(received3.is_some());

        assert_eq!(received1.unwrap().payload()["headline"], "重要新闻");
        assert_eq!(received2.unwrap().payload()["headline"], "重要新闻");
        assert_eq!(received3.unwrap().payload()["headline"], "重要新闻");
    }

    #[tokio::test]
    async fn test_message_ordering() {
        let bus = InMemoryMessageBus::new().await.unwrap();
        let subscriber = bus.subscribe("ordered.messages").await.unwrap();

        // 发布有序消息
        for i in 1..=5 {
            let message = TestHelper::create_test_message(
                "ordered.messages",
                json!({"sequence": i})
            );
            bus.publish(message).await.unwrap();
        }

        // 验证消息顺序
        let mut received_sequence = Vec::new();
        for _ in 0..5 {
            if let Ok(Some(message)) = subscriber.receive_timeout(500).await {
                received_sequence.push(message.payload()["sequence"].as_u64().unwrap());
            }
        }

        assert_eq!(received_sequence, vec![1, 2, 3, 4, 5]);
    }

    #[tokio::test]
    async fn test_subscriber_cleanup() {
        let bus = InMemoryMessageBus::new().await.unwrap();

        // 创建订阅者
        let subscriber = bus.subscribe("cleanup.test").await.unwrap();
        let _subscription_id = subscriber.subscription_id().to_string();

        // 检查订阅者计数
        let initial_count = bus.subscriber_count("cleanup.test").await.unwrap();
        assert_eq!(initial_count, 1);

        // 关闭订阅者
        subscriber.close().await.unwrap();

        // 等待一段时间让清理任务执行
        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
        
        // 检查清理结果
        let final_count = bus.subscriber_count("cleanup.test").await.unwrap_or(0);
        // 清理可能是异步的，所以容忍一定的延迟
        assert!(final_count <= 1, "Expected subscriber count to be 0 or 1, got {}", final_count);
    }

    #[tokio::test]
    async fn test_bus_stats() {
        let bus = InMemoryMessageBus::new().await.unwrap();
        let _subscriber = bus.subscribe("stats.test").await.unwrap();

        // 发布一些消息
        for i in 0..5 {
            let message = TestHelper::create_test_message(
                "stats.test",
                json!({"count": i})
            );
            bus.publish(message).await.unwrap();
        }

        // 检查统计信息
        let stats = bus.stats().await.unwrap();
        assert_eq!(stats.total_published, 5);
        assert_eq!(stats.active_subscribers, 1);
    }

    #[tokio::test]
    async fn test_message_validation() {
        let bus = InMemoryMessageBus::new().await.unwrap();

        // 无效主题
        let invalid_message = Message::new("", json!({"test": "data"}));
        let result = bus.publish(invalid_message).await;
        assert!(result.is_err());

        // 有效消息
        let valid_message = Message::new("valid.topic", json!({"test": "data"}));
        let result = bus.publish(valid_message).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_large_message_handling() {
        let config = MessageBusConfig {
            max_message_size: 1024, // 1KB 限制
            ..Default::default()
        };
        let bus = InMemoryMessageBus::with_config(config).await.unwrap();

        // 创建大消息
        let large_payload = json!({
            "data": "x".repeat(2000) // 超过限制
        });
        let large_message = Message::new("large.message", large_payload);

        // 应该被拒绝
        let result = bus.publish(large_message).await;
        assert!(result.is_err());

        // 小消息应该正常
        let small_message = Message::new("small.message", json!({"data": "small"}));
        let result = bus.publish(small_message).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_concurrent_publishing() {
        let bus = Arc::new(InMemoryMessageBus::new().await.unwrap());
        let subscriber = bus.subscribe("concurrent.*").await.unwrap();

        let message_count = Arc::new(AtomicUsize::new(0));

        // 启动多个并发发布者
        let mut handles = Vec::new();
        for i in 0..10 {
            let bus_clone = bus.clone();
            let count_clone = message_count.clone();
            
            let handle = tokio::spawn(async move {
                for j in 0..10 {
                    let message = TestHelper::create_test_message(
                        &format!("concurrent.publisher{}", i),
                        json!({"id": j})
                    );
                    if bus_clone.publish(message).await.is_ok() {
                        count_clone.fetch_add(1, Ordering::Relaxed);
                    }
                }
            });
            handles.push(handle);
        }

        // 等待所有发布者完成
        for handle in handles {
            handle.await.unwrap();
        }

        // 验证发布的消息数量
        let expected_count = message_count.load(Ordering::Relaxed);
        assert_eq!(expected_count, 100);

        // 验证接收的消息数量
        let received_messages = TestHelper::collect_messages(subscriber, expected_count, 5000).await;
        assert_eq!(received_messages.len(), expected_count);
    }

    #[tokio::test]
    async fn test_bus_close() {
        let bus = InMemoryMessageBus::new().await.unwrap();
        let subscriber = bus.subscribe("close.test").await.unwrap();

        // 正常发布
        let message = TestHelper::create_test_message("close.test", json!({"before": "close"}));
        assert!(bus.publish(message).await.is_ok());
        
        // 先接收消息，确保消息已经发布成功
        let received_before_close = subscriber.receive_timeout(500).await.unwrap();
        assert!(received_before_close.is_some());
        assert_eq!(received_before_close.unwrap().payload()["before"], "close");

        // 关闭消息总线
        assert!(bus.close().await.is_ok());

        // 关闭后发布应该失败
        let message = TestHelper::create_test_message("close.test", json!({"after": "close"}));
        assert!(bus.publish(message).await.is_err());

        // 关闭后订阅应该失败
        assert!(bus.subscribe("new.topic").await.is_err());
    }

    #[tokio::test]
    async fn test_memory_pressure() {
        // 测试内存压力场景
        let config = MessageBusConfig {
            buffer_size: 50, // 更小的缓冲区
            ..Default::default()
        };
        let bus = InMemoryMessageBus::with_config(config).await.unwrap();

        // 创建订阅者
        let subscriber = bus.subscribe("pressure.test").await.unwrap();

        // 使用超时控制防止测试卡死
        let test_future = async {
            let mut published_count = 0;
            let mut consumed_count = 0;
            
            // 发布消息，但不要发布太多以免卡死
            for i in 0..200 {
                let message = TestHelper::create_test_message(
                    "pressure.test",
                    json!({"id": i})
                );
                
                // 发布可能因缓冲区满而失败，这是正常的
                match tokio::time::timeout(
                    tokio::time::Duration::from_millis(10), 
                    bus.publish(message)
                ).await {
                    Ok(Ok(())) => published_count += 1,
                    Ok(Err(_)) | Err(_) => {
                        // 如果发布失败或超时，尝试消费一些消息
                        for _ in 0..5 {
                            if let Ok(Some(_)) = subscriber.try_receive().await {
                                consumed_count += 1;
                            } else {
                                break;
                            }
                        }
                        tokio::task::yield_now().await;
                    }
                }
                
                // 定期消费消息
                if i % 20 == 0 {
                    for _ in 0..10 {
                        match subscriber.try_receive().await {
                            Ok(Some(_)) => consumed_count += 1,
                            _ => break,
                        }
                    }
                    tokio::task::yield_now().await;
                }
            }

            // 消费剩余的消息
            for _ in 0..100 {
                match tokio::time::timeout(
                    tokio::time::Duration::from_millis(10),
                    subscriber.receive()
                ).await {
                    Ok(Ok(Some(_))) => consumed_count += 1,
                    _ => break,
                }
            }

            // 验证至少发布了一些消息
            assert!(published_count > 0, "应该至少发布了一些消息");
            tracing::info!("在内存压力测试中成功发布了 {} 条消息，消费了 {} 条消息", published_count, consumed_count);
        };
        
        // 设置整个测试的超时时间，防止卡死
        tokio::time::timeout(
            tokio::time::Duration::from_secs(10),
            test_future
        ).await.expect("内存压力测试超时");
    }
}

/// 性能测试模块
#[cfg(test)]
mod performance_tests {
    use super::*;
    use std::time::Instant;

    #[tokio::test]
    async fn bench_message_creation() {
        let start = Instant::now();
        let count = 10000;

        for i in 0..count {
            let _message = TestHelper::create_test_message(
                "bench.topic",
                json!({"id": i, "data": "test data"})
            );
        }

        let duration = start.elapsed();
        let messages_per_second = count as f64 / duration.as_secs_f64();
        
        tracing::info!(
            "消息创建性能: {} 条消息/秒 (总共 {} 条消息，耗时 {:?})",
            messages_per_second as u64,
            count,
            duration
        );

        // 应该能够每秒创建至少 100,000 条消息
        assert!(messages_per_second > 100_000.0);
    }

    #[tokio::test]
    async fn bench_publish_subscribe() {
        let bus = InMemoryMessageBus::new().await.unwrap();
        let subscriber = bus.subscribe("bench.*").await.unwrap();

        let start = Instant::now();
        let count = 1000;

        // 发布消息
        for i in 0..count {
            let message = TestHelper::create_test_message(
                "bench.topic",
                json!({"id": i})
            );
            bus.publish(message).await.unwrap();
        }

        // 接收消息
        let mut received_count = 0;
        while received_count < count {
            match timeout(Duration::from_millis(100), subscriber.receive()).await {
                Ok(Ok(Some(_))) => {
                    received_count += 1;
                }
                Ok(Ok(None)) => break,
                Ok(Err(_)) => break,
                Err(_) => break, // 超时
            }
        }

        let duration = start.elapsed();
        let messages_per_second = count as f64 / duration.as_secs_f64();

        tracing::info!(
            "发布订阅性能: {} 条消息/秒 (总共 {} 条消息，耗时 {:?})",
            messages_per_second as u64,
            count,
            duration
        );

        assert_eq!(received_count, count);
        // 应该能够每秒处理至少 10,000 条消息
        assert!(messages_per_second > 10_000.0);
    }
}